package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;

/**
 * 用例6 Service
 */
@Service
public class ConsumerLeakService {
    private Consumer consumer;

    public String query(String key) {
        // 临时查询创建 Consumer，未关闭
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(KafkaConfig.getConsumerProps());
        consumer.subscribe(Collections.singletonList("topic1"));
        // ... 查询逻辑 ...
        // 未关闭 consumer
        return "result";
    }

    public void startThread() {
        new Thread(() -> {
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(KafkaConfig.getConsumerProps());
            consumer.subscribe(Collections.singletonList("topic2"));
            // ... 消费逻辑 ...
            // 线程结束未关闭 consumer
        }).start();
    }

    public void consume() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(KafkaConfig.getConsumerProps());
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                // ... 消费逻辑 ...
                if (records.isEmpty()) {
                    throw new RuntimeException("no data");
                }
            }
        } catch (Exception e) {
            // 异常时未关闭 consumer
        }
    }

    public void changePartition(int partition) {
        // 动态调整分区时创建新 Consumer，未关闭旧实例
        consumer = new KafkaConsumer<>(KafkaConfig.getConsumerProps());
        consumer.assign(Arrays.asList(new TopicPartition("topic4", partition)));
        // 未关闭之前的 consumer
    }

    public void monitor() {
        // 临时监控任务创建 Consumer，未关闭
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(KafkaConfig.getConsumerProps());
        consumer.subscribe(Collections.singletonList("topic5"));
        // ... 监控逻辑 ...
        // 未关闭 consumer
    }
}
